package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.dataCenter.DiskDataCenter;
import com.example.mq.mqserver.dataCenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 用此类表示虚拟机
 * 每个虚拟机都管理者自己的: 交换机 队列 绑定 消息 数据
 * VirtualHost 作为业务逻辑类的整合 需要对抛出的异常做处理
 */
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    // 操作队列的锁对象
    private final Object queueLocker = new Object();

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String virtualHostName){
        this.virtualHostName = virtualHostName;
        //进行硬盘数据的初始化:建库建表
        diskDataCenter.init();

        //第一次启动server:不需要恢复内存数据
        //server启动已久:恢复内存数据
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败.");
        }
    }

    //创建交换机
    //交换机存在:不创建
    //交换机不存在:创建
    //返回true 创建成功.false 为失败
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType,
                                   boolean durable, boolean autoDelete, Map<String,Object> arguments){
        //为交换机加上虚拟机的前缀
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker){
                //查看内存中交换机是否存在
                Exchange existExchange = memoryDataCenter.getExchange(exchangeName);
                if(existExchange != null){
                    //交换机已经存在
                    System.out.println("[VirtualHost] 交换机已经存在. exchangeName = "+exchangeName);
                    return true;
                }

                //交换机不存在.设置交换机参数
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);

                //如果需要进行持久化:将交换机对象写入硬盘中
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }
                //交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成.exchangeName = "+exchangeName);
                //上述逻辑先写硬盘再写内存 ----  写入硬盘操作负责,如果出现错误内存也可以不用添加。否则内存成功,写入硬盘失败，会需要再次删除内存数据.
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败.exchangeName = "+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除对应的交换机
    public boolean exchangeDelete(String exchangeName){
        try {
            synchronized (exchangeLocker){
                //添加虚拟机前缀
                exchangeName = virtualHostName + exchangeName;
                //查看对应的交换机是否存在
                Exchange existExchange = memoryDataCenter.getExchange(exchangeName);
                if(existExchange == null){
                    //交换机不存在
                    throw new MqException("[VirtualHost] 交换机不存在,无法进行删除.exchangeName ="+exchangeName);
                }
                //查看是否持久化保存 是:删除硬盘中的交换机数据
                if(existExchange.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功.exchangeName ="+exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机删除成功.exchangeName ="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDeclare(String queueName,boolean durable,
                                boolean exclusive,boolean autoDelete,Map<String,Object> arguments){
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker){
                //查看队列是否存在
                MSGQueue existQueue = memoryDataCenter.getQueue(queueName);
                if(existQueue != null){
                    System.out.println("[VirtualHost] 队列已经存在. queueName = "+queueName);
                    return true;
                }

                //不存在就创建队列准备写入
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);

                //写入硬盘
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }
                //写入内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功.queueName"+queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列创建失败.queueName"+queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 根据队列名字, 查询下当前的队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在.无法删除. queueName=" + queueName);
                }
                // 2. 删除硬盘数据
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                // 3. 删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功. queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除队列失败.queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //绑定交换机和队列
    public boolean queueBind(String queueName,String exchangeName,String bindingKey){
        //添加虚拟机前缀
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //查看绑定是否存在
                    Binding existBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existBinding != null){
                        throw new MqException("[VirtualHost] 绑定已经存在 exchangeName = " + exchangeName
                                +",queueName =" + queueName);
                    }

                    //验证binding是否合法
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey非法.bindingKey = "+bindingKey);
                    }

                    //构造Binding对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);

                    //获取对应的交换机和队列 不存在的话无法创建绑定
                    MSGQueue existQueue = memoryDataCenter.getQueue(queueName);
                    if(existQueue == null){
                        throw new MqException("[VirtualHost] 队列不存在 queueName ="+queueName);
                    }
                    Exchange existExchange = memoryDataCenter.getExchange(exchangeName);
                    if(existExchange == null){
                        throw new MqException("[VirtualHost] 交换机不存在 exchangeName ="+exchangeName);
                    }

                    //绑定持久化保存
                    if(existExchange.isDurable() && existQueue.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }

                    //写入内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功.exchangeName = "+ exchangeName + ",queueName = "+queueName);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 绑定创建失败.exchangeName = "+ exchangeName + ",queueName = "+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除交换机和队列之间的绑定关系
    public boolean queueUnbind(String queueName,String exchangeName){
        //添加前缀
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    //查看绑定是否存在
                    Binding existBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existBinding == null){
                        throw new MqException("[VirtualHost] 绑定不存在.无法删除 exchangeName = "
                                +exchangeName + ",queueName ="+queueName);
                    }
                    //无论绑定是否持久化 都尝试在硬盘上删除绑定.及时绑定不存在,删除也不会存在副作用
                    diskDataCenter.deleteBinding(existBinding);
                    // 删除内存上的数据
                    memoryDataCenter.deleteBinding(existBinding);
                    System.out.println("[VirtualHost] 删除绑定成功.exchangeName = "
                            +exchangeName + ",queueName ="+queueName);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 删除绑定失败.exchangeName = "
                    +exchangeName + ",queueName ="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 发送消息到指定的交换机/队列中.
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
        try {
            //添加前缀
            exchangeName = virtualHostName + exchangeName;
            //检查 routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey不合法. routingKey = "+routingKey);
            }
            //查看交换机是否存在
            Exchange existExchange = memoryDataCenter.getExchange(exchangeName);
            if(existExchange == null){
                throw new MqException("[VirtualHost] 交换机不存在. exchangeName = "+exchangeName);
            }

            //判断交换机的类型
            if(existExchange.getType() == ExchangeType.DIRECT){
                //按照直接交换机的方式进行消息发送
                //以 routingKey 作为队列的名字, 直接把消息写入指定的队列中. 此时可以无视绑定关系
                String queueName = virtualHostName + routingKey;

                //构造消息对象
                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                //查找该队列对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost] 队列不存在. queueName = "+queueName);
                }

                //队列存在 直接往队列里发送消息
                sendMessage(queue,message);
                return true;
            }else {
                //按照 fanout 和 topic 的交换机类型进行转发
                //找到该交换机的所有绑定 并遍历这些绑定对象
                ConcurrentHashMap<String,Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String,Binding> entry : bindingsMap.entrySet() ){
                    //获取绑定对象 判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue == null){
                        //一个绑定有很多队列.不能因为一个队列产生异常就停止
                        System.out.println("[VirtualHost] 发送消息时,队列不存在. queueName = "+binding.getQueueName());
                        continue;
                    }
                    //构造消息对象
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    //判定这个消息是否能转发给该队列:
                    //如果是fanout 类型,每个队列都发
                    //如果是 topic 类型,需要判定 bindingKey 和 routingKey 是不是匹配.
                    if(!router.route(existExchange.getType(),binding,message)){
                        continue;
                    }
                    //转发消息给队列
                    sendMessage(queue,message);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 发送消息失败");
            e.printStackTrace();
            return false;
        }

    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 此处发送消息, 就是把消息写入到 硬盘 和 内存 上.
        int deliverMode = message.getDeliverMode();
        // deliverMode 为 1 , 不持久化. deliverMode 为 2 表示持久化.
        if (deliverMode == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        // 写入内存
        memoryDataCenter.sendMessage(queue, message);

        //消息添加完毕,需要提醒消费者消费消息
        consumerManager.notifyConsume(queue.getName());
    }

    // 订阅消息.
    // 添加一个队列的订阅者, 当队列收到消息之后, 就要把消息推送给对应的订阅者.
    // consumerTag: 消费者的身份标识
    // autoAck: 消息被消费完成后, 应答的方式. 为 true 自动应答. 为 false 手动应答.
    // consumer: 是一个回调函数. 此处类型设定成函数式接口. 这样后续调用 basicConsume 并且传实参的时候, 就可以写作 lambda 样子了.
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        //构造一个 ConsumerEnv 对象,把对应的队列找到.再把 Consumer对象放进队列之中
        queueName = virtualHostName + queueName;
        try{
            //添加一个消费者
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] 消息推送给对应订阅者成功.queueName ="+queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 消息推送给对应订阅者失败.queueName ="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //消息被消费后消费者应答 ---- 表示消息被消费 ---- 消息可以被删除
    public boolean basicAck(String queueName, String messageId){
        queueName = virtualHostName + queueName;
        try {
            // 1. 获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 要确认的消息不存在. messageId=" + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 要确认的队列不存在. queueName=" + queueName);
            }
            // 2. 删除硬盘上的数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            // 3. 删除消息中心中的数据
            memoryDataCenter.removeMessage(messageId);
            // 4. 删除待确认的集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck 成功. 消息被成功确认. queueName=" + queueName
                    + ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck 失败. 消息确认失败. queueName=" + queueName
                    + ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
